package job;


import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class Vehicle_d {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        tenv.executeSql("CREATE TABLE products (\n" +
                "                vin STRING ,\n" +
                "                batteryLevel DECIMAL(10, 2) ,\n" +
                "                batteryTemp DECIMAL(10, 2),\n" +
                "                motorTemp DECIMAL(10, 2),\n" +
                "                batteryds INT,\n" +
                "                batterysh DECIMAL(10, 2),\n" +
                "                startdl DECIMAL(10, 2),\n" +
                "                enddl DECIMAL(10, 2),\n" +
                "                bcrl INT,\n" +
                "                timestamp1 BIGINT,\n" +
                "                PRIMARY KEY(vin) NOT ENFORCED\n" +
                ") WITH (\n" +
                "                'connector' = 'mysql-cdc',\n" +
                "                'hostname' = 'node00',\n" +
                "                'port' = '3306',\n" +
                "                'username' = 'root',\n" +
                "                'password' = 'root',\n" +
                "                'database-name' = 'test',\n" +
                "                'table-name' = 'battery1'\n" +
                "        );");

        tenv.executeSql("CREATE TABLE products1 (\n" +
                "                vin STRING ,\n" +
                "                timestamp1 BIGINT,\n" +
                "                longitude DECIMAL(10, 2),\n" +
                "                latitude DECIMAL(10, 2),\n" +
                "                speed DECIMAL(10, 2),\n" +
                "                mileage DECIMAL(10, 2),\n" +
                "                batteryLevel DECIMAL(10, 2),\n" +
                "                battertTemp DECIMAL(10, 2),\n" +
                "                motorTemp DECIMAL(10, 2),\n" +
                "                chargingStatus string ,\n" +
                "                energyConsumption DECIMAL(10, 2),\n" +
                "                PRIMARY KEY(vin) NOT ENFORCED\n" +
                ") WITH (\n" +
                "                'connector' = 'mysql-cdc',\n" +
                "                'hostname' = 'node00',\n" +
                "                'port' = '3306',\n" +
                "                'username' = 'root',\n" +
                "                'password' = 'root',\n" +
                "                'database-name' = 'test',\n" +
                "                'table-name' = 'VehicleData'\n" +
                "        );");


        Table table = tenv.sqlQuery("SELECT * FROM products");
//        tenv.toChangelogStream(table).print();

        Table table1 = tenv.sqlQuery("SELECT * FROM products1");
//        tenv.toChangelogStream(table1).print();

//        实时在线车辆数、平均电池健康度、车辆地理分布热力图
        Table table2 = tenv.sqlQuery("WITH BatteryHealth AS (\n" +
                "    SELECT\n" +
                "        b.vin,\n" +
                "        (b.batteryds * b.batterysh / (b.enddl - b.startdl) / b.bcrl * 100) AS soh\n" +
                "    FROM\n" +
                "        products b\n" +
                "    WHERE\n" +
                "        b.enddl IS NOT NULL AND b.startdl IS NOT NULL AND b.bcrl IS NOT NULL AND b.batterysh IS NOT NULL AND b.batteryds IS NOT NULL\n" +
                "),\n" +
                "     LatestVehicleData AS (\n" +
                "         SELECT\n" +
                "             v.vin,\n" +
                "             v.longitude,\n" +
                "             v.latitude,\n" +
                "             ROW_NUMBER() OVER (PARTITION BY v.vin ORDER BY v.timestamp1 DESC) AS rn\n" +
                "         FROM\n" +
                "             products1 v\n" +
                "     )\n" +
                "SELECT\n" +
                "    COUNT(DISTINCT bh.vin) AS vehicle_count,\n" +
                "    AVG(bh.soh) AS avg_soh,\n" +
                "    MAX(CASE WHEN lvd.rn = 1 THEN lvd.longitude END) AS latest_longitude,\n" +
                "    MAX(CASE WHEN lvd.rn = 1 THEN lvd.latitude END) AS latest_latitude\n" +
                "FROM\n" +
                "    BatteryHealth bh\n" +
                "        JOIN\n" +
                "    LatestVehicleData lvd ON bh.vin = lvd.vin\n" +
                "WHERE\n" +
                "        lvd.rn = 1;");
        tenv.toChangelogStream(table2).print();

//        tenv.toChangelogStream(table2).addSink(
//                JdbcSink.sink(
//                        "insert into rlt (vehicle_count, avg_soh, latest_longitude, latest_latitude) values (?, ?, ?, ?)",
//                        (statement, book) -> {
//                            statement.setObject(1, book.getField("vehicle_count"));
//                            statement.setObject(2, book.getField("avg_soh"));
//                            statement.setObject(3, book.getField("latest_longitude"));
//                            statement.setObject(4, book.getField("latest_latitude"));
//                        },
//                        JdbcExecutionOptions.builder()
//                                .withBatchSize(1000)
//                                .withBatchIntervalMs(200)
//                                .withMaxRetries(5)
//                                .build(),
//                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                                .withUrl("jdbc:mysql://node00:3306/test")
//                                .withUsername("root")
//                                .withPassword("root")
//                                .build()
//                ));

        env.execute("xxxxxx");
    }
}
